package com.hjs.homework.window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Iterator;
import java.util.Random;

public class EventWindowJava {
    public static void main(String[] args) throws Exception {
        String host = "bogon";
        int port = 7777;

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> streamSource = env.socketTextStream(host, port);

        SingleOutputStreamOperator<Tuple3<String, String, String>> maped = streamSource.map(new MapFunction<String, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> map(String s) throws Exception {
                SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                long l = System.currentTimeMillis();
                String dataTime = simpleDateFormat.format(l);
                Random random = new Random();
                return new Tuple3<String, String, String>(s, dataTime, String.valueOf(random.nextInt(5)));
            }
        });

        KeyedStream<Tuple3<String, String, String>, String> keyed = maped.keyBy(value -> value.f0);

        WindowedStream<Tuple3<String, String, String>, String, GlobalWindow> countWindow = keyed.countWindow(3,1);

        SingleOutputStreamOperator<String> apply = countWindow.apply(new WindowFunction<Tuple3<String, String, String>, String, String, GlobalWindow>() {
            @Override
            public void apply(String s, GlobalWindow globalWindow, Iterable<Tuple3<String, String, String>> iterable, Collector<String> collector) throws Exception {
                Iterator<Tuple3<String, String, String>> iterator = iterable.iterator();
                StringBuilder sb = new StringBuilder();
                while (iterator.hasNext()){
                    Tuple3<String, String, String> next = iterator.next();
                    sb.append(next.f0 + "..." + next.f1 + "..." + next.f2);
                }
                collector.collect(sb.toString());
            }
        });

        apply.print();

        env.execute();
    }
}
